/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package bigo.lib;

import bigo.data.ArrayVector;
import bigo.data.HashVector;
import bigo.data.Matrix;
import bigo.data.MatrixMeta;
import bigo.data.Vector;
import java.io.BufferedReader;
import java.io.File;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.TreeMap;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileContext;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.io.compress.CompressionCodecFactory;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.JobContext;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.hadoop.mapreduce.lib.input.LineRecordReader;

/**
 *
 * @author songliu
 */
public class TextMatrixInputFormat extends FileInputFormat<IntWritable, Vector.Wrapper> {

    @Override
    public RecordReader<IntWritable, Vector.Wrapper> createRecordReader(InputSplit is, TaskAttemptContext tac) throws IOException, InterruptedException {
        return new Reader();
    }

    public class Reader extends RecordReader<IntWritable, Vector.Wrapper> {

        private int rowNo;
        private Vector v;
        private LineRecordReader r = new LineRecordReader();
        private MatrixMeta meta;

        @Override
        public void initialize(InputSplit is, TaskAttemptContext tac) throws IOException, InterruptedException {
            r.initialize(is, tac);
            FileSplit fs = (FileSplit) is;
            String str = tac.getConfiguration().get(fs.getPath().getParent().getName());
            //obtain the matrix meta using its path as its key.
            meta = new MatrixMeta(str);
        }

        @Override
        public boolean nextKeyValue() throws IOException, InterruptedException {
            boolean res = r.nextKeyValue();
            if (res) {
                rowNo = Integer.parseInt(
                        r.getCurrentValue().toString().split("\\s")[0]);
                v = readLine(r.getCurrentValue().toString(), meta);
                return true;
            }
            return false;

        }

        @Override
        public IntWritable getCurrentKey() throws IOException, InterruptedException {
            return new IntWritable(rowNo);
        }

        @Override
        public Vector.Wrapper getCurrentValue() throws IOException, InterruptedException {
            return new Vector.Wrapper(v);
        }

        @Override
        public float getProgress() throws IOException, InterruptedException {
            return r.getProgress();
        }

        @Override
        public void close() throws IOException {
            r.close();
        }
    }

    @Override
    protected boolean isSplitable(JobContext context, Path file) {
        CompressionCodec codec = new CompressionCodecFactory(context.getConfiguration()).getCodec(file);
        return codec == null;
    }

    public static void addInputMatrix(Job job, MatrixMeta meta) throws IOException {
        //store the matrix meta using its path as its key
        job.getConfiguration().set(meta.path, meta.toString());
        FileInputFormat.addInputPath(job, new Path(meta.path));
    }

    public static void main(String[] args) throws IOException, InterruptedException {
        ProcessBuilder pb = new ProcessBuilder(
                "sh", "-c", "cat A/* | sort --key=1,2 -n > test.txt");

        BufferedReader br = new BufferedReader(
                new InputStreamReader(pb.start().getInputStream()));
        String line = br.readLine();
        while (line != null) {
            System.out.println(line);
            line = br.readLine();
        }
    }

    public static Matrix loadMatrix(Configuration conf, MatrixMeta meta) throws IOException {
//        //TODO: Sort the matrix by row
//        ProcessBuilder pb = new ProcessBuilder(
//                "sh", "-c", "cat " + meta.path + "/* | sort --key=1,2 -n");

        TreeMap<Integer, Vector> matrixStore = new TreeMap<Integer, Vector>();
        for (File f : new File(meta.path).listFiles()) {
            if (f.getName().startsWith("_") || f.getName().startsWith(".")) {
                continue;
            }

            BufferedReader br = new BufferedReader(
                    new InputStreamReader(
                    FileContext.getLocalFSFileContext().open(
                    new Path(f.getPath()))));

            String line = br.readLine();
            while (line != null) {
                int rowIndex = Integer.parseInt(line.substring(0, line.indexOf(" ")));
                Vector v = readLine(line, meta);
                matrixStore.put(rowIndex, v);
                line = br.readLine();
            }
            br.close();
        }
        
        Matrix ret = new Matrix();
        for(Vector v:matrixStore.values()){
            ret.addRowVector(v);
        }
        return ret;
    }

    public static Vector readLine(String line, MatrixMeta meta) {

        String vectorStr = line.split("\\s")[1];
        Vector v = null;
        if (vectorStr.contains(":")) {
            //sparse format
            v = new HashVector(vectorStr.split(",").length, meta.numColumns);
            for (String s : vectorStr.split(",")) {
                int col = Integer.parseInt(s.split(":")[0]);
                Float val = Float.parseFloat(s.split(":")[1]);
                v.put(col, val);
            }
        } else {
            v = new ArrayVector(vectorStr.split(",").length);
            //dense format
            int i = 0;
            for (String s : vectorStr.split(",")) {
                v.put(i, Float.parseFloat(s));
                i++;
            }
        }
        return v;
    }
}
